package com.atguigu.state;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

public class Flink01_Operator_ListState {
    public static void main(String[] args) {
        Configuration configuration = new Configuration();
        configuration.setInteger("rqest.port",10000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setParallelism(2);

        //只是该应用的状态，如果重启程序的话就又是另一个应用了，所以该状态就可能不可用了！
        //每三秒做一次快照
        env.enableCheckpointing(3000);


        DataStreamSource<String> ds = env.socketTextStream("hadoop162", 9999);
        ds
                //自定义flatMap函数
                .flatMap(new MyFlatMapFunction())
                //打印输出结果
                .print();

        //在这里我们做一个实验，看看应用重启的时候，会不会恢复状态
        ds.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                if (value.contains("x")) {
                    throw new RuntimeException("自定义异常，看看重启之后状态是否恢复！");
                }
            }
        });


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static class MyFlatMapFunction implements CheckpointedFunction, FlatMapFunction<String,String>{
        //创建数据结构来存放临时的数据
        List<String> list = new ArrayList();
        private ListState<String> wordState;
        //我们需要定义一个中间变量进行临时存储每个数据的状态，然后在将该数据结构中的数据传给wordState
        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            //在这里进行flatmap的逻辑处理，并将结果保存到状态中
            String[] datas = value.split(" ");
            for (String data : datas) {
                //将结果保存在临时状态中
                list.add(data);
                out.collect(list.toString());
            }
        }

        //一个周期执行一次
        //做状态保存用的
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            //这里进行周期性的状态保存
            //先将原先的状态删除，然后存入最新的状态
            /*wordState.clear();
            wordState.add(list.toString());*/
            //以上两步可以简化为一步
            wordState.update(list);
        }

        //只有程序启动，重启的时候会启动一次
        //做恢复状态用的
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {

            System.out.println("MyFlatMapFunction.initializeState"+"初始化执行！！");
            //在状态中保存什么值，就在这里定义该值的类型
            wordState = context
                    .getOperatorStateStore()
                    .getListState(new ListStateDescriptor<String>("wordState", String.class));

            //重启的时候将之前状态返回
            Iterable<String> states = wordState.get();
            for (String state : states) {
                list.add(state);
            }
        }
    }
}
